package com.tensai.study.java.newconcept.typereactiveprogram.reactor.flux;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.function.Function;

/**
 * @author Tensai
 */
@Slf4j
public class ConcatMapTest {
    Function<Flux<String>, Flux<String>> function =
            flux -> flux.map(String::toUpperCase).delayElements(Duration.ofMillis(200));

    Function<Flux<Flux<String>>, Flux<String>> concatMapFunction = flux ->
            flux.concatMap(flux1 -> function.apply(flux1));

    Function<Flux<Flux<String>>, Flux<String>> flatMapFunction = flux ->
            flux.flatMap(flux1 -> function.apply(flux1));


    @Test
    public void concatMap() {
        common(concatMapFunction);
    }

    /**
     * <li>{@link Flux#concatMap(Function)} 是有序的</li>
     * <li>{@link Flux#flatMap(Function)} 是无序的 </li>
     */
    @Test
    public void concatMapDiffFlatMap() {
        log.info("concatMap");
        concatMap();
        log.info("flatMap");
        common(flatMapFunction);
    }

    private void common(Function<Flux<Flux<String>>, Flux<String>> function) {
        Flux<String> stringFlux1 = Flux.just("a", "b", "c", "d", "e", "f", "g", "h", "i");
        Flux<Flux<String>> stringFlux2 = stringFlux1.window(2);

        function.apply(stringFlux2).subscribe(x -> log.info("->" + x));
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
